package com.dji.sample.manage.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.dji.sample.common.constant.CommonConstant;
import com.dji.sample.common.model.CacheModel;
import com.dji.sample.common.model.StoreCache;
import com.dji.sample.component.mqtt.model.CommonTopicReceiver;
import com.dji.sample.component.mqtt.service.IMqttMessageGateway;
import com.dji.sample.component.websocket.config.ConcurrentWebSocketSession;
import com.dji.sample.component.websocket.model.CustomWebSocketMessage;
import com.dji.sample.manage.model.dto.DeviceDTO;
import com.dji.sample.manage.model.dto.TelemetryDTO;
import com.dji.sample.manage.model.enums.DeviceDomainEnum;
import com.dji.sample.manage.model.enums.TimeHeartbeatEnum;
import com.dji.sample.manage.model.receiver.BrokerPositionReceiver;
import com.dji.sample.manage.model.receiver.DevicePositionReceiver;
import com.dji.sample.manage.model.receiver.PositionItemReceiver;
import com.dji.sample.manage.service.DevicePositionService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.Collection;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author yjt
 * @date 2022-09-08
 * @description:
 */
@Service
@Slf4j
@Transactional
public class DevicesPositionOSDServiceImpl extends AbstractTSAService implements InitializingBean {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    DevicePositionService devicePositionService;

    private StoreCache storeCache;

    @Autowired
    private IMqttMessageGateway messageGateway;

    @Autowired
    private ObjectMapper mapper;

    public DevicesPositionOSDServiceImpl() {
        super(null);
    }

    @Override
    public void pushTelemetryData(Integer type, Collection<ConcurrentWebSocketSession> sessions, CustomWebSocketMessage<TelemetryDTO> message, Object Object) {

    }

    @Override
    public void handleOSD(Integer type, CommonTopicReceiver receiver, DeviceDTO device, Collection<ConcurrentWebSocketSession> webSessions, CustomWebSocketMessage<TelemetryDTO> wsMessage) {
        //log.info("心跳设备信息{}", device.toString());
        if (!device.getDomain().equals(DeviceDomainEnum.SUB_DEVICE.getDesc())) {
            log.info("心跳设备信息{}", device.toString());
            return;
        }
        if (!StringUtils.hasText(device.getDeviceSn())) {
            log.info("======设备序列号为空======{}", receiver.getData());
            return;
        }
        Map<String, Object> data = (Map<String, Object>) receiver.getData();
        if (Objects.isNull(data.get("latitude")) || Objects.isNull(data.get("longitude"))) {
            log.debug("======无经纬度数据======{}", receiver.getData());
            return;
        }
        String height = Objects.isNull(data.get("elevation")) ? "0" : String.valueOf(data.get("elevation"));
        String latitude = Objects.isNull(data.get("latitude")) ? "0" : String.valueOf(data.get("latitude"));
        String longitude = Objects.isNull(data.get("longitude")) ? "0" : String.valueOf(data.get("longitude"));
        if ("0".equals(latitude) || "0".equals(longitude) || "0.0".equals(latitude) || "0.0".equals(longitude)) {
            log.debug("======无经纬度数据======{}", receiver.getData());
            return;
        }
        DevicePositionReceiver devicePositionReceiver = new DevicePositionReceiver(height, latitude, longitude);
        devicePositionReceiver.setTid(receiver.getTid());
        devicePositionReceiver.setBid(receiver.getBid());
        devicePositionReceiver.setTimestamp(receiver.getTimestamp());
        devicePositionReceiver.setGateway(receiver.getGateway());
        devicePositionReceiver.setData(receiver.getData().toString());
        devicePositionReceiver.setDeviceSn(device.getDeviceSn());

        //broker节点订阅处理火情信息
        pushPosition(devicePositionReceiver.getPositionItemReceiver(),device.getDeviceSn(),receiver.getTimestamp());
        //缓存10s 无人机信息查询实时点位需要 json存储
        String heartKey = String.format("%s%s", "device:osd:position:heartbeat:", device.getDeviceSn());
        redisTemplate.opsForValue().set(heartKey, JSONObject.toJSONString(receiver), 5, TimeUnit.SECONDS);
        //10S 入库一次本地缓存 + redis缓存 命中本地则不走redis缓存 无人机下线时删除本地缓存 redis.hash(key)
        String key = String.format("%s%s", "device:osd:position:", device.getDeviceSn());
        long newTime = System.currentTimeMillis() / TimeHeartbeatEnum.THOUSAND.getVal();
        if (storeCache.containsKey(key)) {
            Object val = storeCache.get(key);
            if (newTime - Long.parseLong(val.toString()) >= 10) {
                //通过 允许入库
                updateCache(key, newTime);
                devicePositionService.save(devicePositionReceiver);
            }
        } else if (Boolean.TRUE.equals(redisTemplate.hasKey(key))) {
            Object val = redisTemplate.opsForValue().get(key);
            if (Objects.nonNull(val) && (newTime - Long.parseLong(val.toString()) >= 10)) {
                //通过
                updateCache(key, newTime);
                devicePositionService.save(devicePositionReceiver);
            }
        } else {
            updateCache(key, newTime);
            devicePositionService.save(devicePositionReceiver);
        }
    }

    public void updateCache(String key, Long newTime) {
        storeCache.put(key, newTime);
        redisTemplate.opsForValue().set(key, newTime, 30, TimeUnit.SECONDS);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        storeCache = CacheModel.get("devicesPositionOSDServiceImpl").name("handleOSD");
    }

    public void pushPosition(PositionItemReceiver positionItemReceiver, String deviceSn, Long timestamp) {
        try {
            BrokerPositionReceiver brokerPositionReceiver = new BrokerPositionReceiver();
            brokerPositionReceiver.setElevation(positionItemReceiver.getHeight());
            brokerPositionReceiver.setLongitude(positionItemReceiver.getLongitude());
            brokerPositionReceiver.setLatitude(positionItemReceiver.getLatitude());
            brokerPositionReceiver.setDeviceSn(deviceSn);
            brokerPositionReceiver.setTimestamp(timestamp);
            messageGateway.publish(CommonConstant.MQ_DEVICE_POS_TOPIC, mapper.writeValueAsBytes(brokerPositionReceiver));
        } catch (JsonProcessingException e) {
            log.error("发送消息异常", e);
        }
    }

}
